package com.raven.play.stream.test.scaladsl.tutorial

import akka.actor.ActorSystem
import akka.stream.{FlowShape, SinkShape}
import akka.stream.scaladsl.{Broadcast, Flow, GraphDSL, Keep, Sink, Source}

import scala.concurrent.Future
import scala.util.{Failure, Success}

object $12GraphMaterializedValues extends App {

  implicit val system = ActorSystem("GraphMaterializedValues")
  import system.dispatcher

  val wordSource = Source(List("Akka", "is", "awesome", "rock", "the", "jvm"))
  val printer = Sink.foreach[String](println)
  val counter = Sink.fold[Int, String](0)((count, _) => count + 1)

  /**
   * A   composite component(sink)
   * prints out all strings which are lowercase
   * COUNTS the strings that are short(< 5 chars)
   */

  // step 1
  val complexWordSink = Sink.fromGraph(
    //    GraphDSL.create() { implicit builder =>
    GraphDSL.create() { implicit builder =>
      import GraphDSL.Implicits._
      // step 2 - SHAPES
      val broadcast = builder.add(Broadcast[String](2))
      val lowercaseFilter = builder.add(Flow[String].filter(word => word == word.toLowerCase))
      val shortStringFilter = builder.add(Flow[String].filter(_.length < 5))
      // step 3 - connections
      broadcast ~> lowercaseFilter ~> printer
      broadcast ~> shortStringFilter ~> counter
      // step 4 - the shape
      SinkShape(broadcast.in)
    })

  val complexWordSink2 = Sink.fromGraph(
    GraphDSL.create(counter) { implicit builder =>
      counterShape =>
        import GraphDSL.Implicits._
        // step 2 - SHAPES
        val broadcast = builder.add(Broadcast[String](2))
        val lowercaseFilter = builder.add(Flow[String].filter(word => word == word.toLowerCase))
        val shortStringFilter = builder.add(Flow[String].filter(_.length < 5))
        // step 3 - connections
        broadcast ~> lowercaseFilter ~> printer
        broadcast ~> shortStringFilter ~> counterShape
        // step 4 - the shape
        SinkShape(broadcast.in)
    })

  val complexWordSink3 = Sink.fromGraph(
    //    GraphDSL.create() { implicit builder =>
    GraphDSL.create(printer, counter)((printerMatValue, counterMatValue) => counterMatValue) { implicit builder =>
      (printerShape, counterShape) =>
        import GraphDSL.Implicits._
        // step 2 - SHAPES
        val broadcast = builder.add(Broadcast[String](2))
        val lowercaseFilter = builder.add(Flow[String].filter(word => word == word.toLowerCase))
        val shortStringFilter = builder.add(Flow[String].filter(_.length < 5))
        // step 3 - connections
        broadcast ~> lowercaseFilter ~> printerShape
        broadcast ~> shortStringFilter ~> counterShape
        // step 4 - the shape
        SinkShape(broadcast.in)
    }
  )

//  val shortStringsCountFuture = wordSource.toMat(complexWordSink3)(Keep.right).run()
//  shortStringsCountFuture.onComplete {
//    case Success(count) => println(s"The total number of short strings is: $count")
//    case Failure(exception) => println(s"The count of short strings failed: $exception")
//  }

  /**
   * Exercise
  */
  def enhanceFlow[A, B](flow: Flow[A, B, _]): Flow[A, B,Future[Int]]={
    val counterSink = Sink.fold[Int, B](0)((count,_) => count + 1)
    Flow.fromGraph(
      GraphDSL.create(counterSink) { implicit builder =>
        counterSinkShape =>
          import GraphDSL.Implicits._
          val broadcast = builder.add(Broadcast[B](2))
          val originalFlowShape = builder.add(flow)

          originalFlowShape ~> broadcast ~> counterSinkShape

          FlowShape(originalFlowShape.in, broadcast.out(1))

      }
    )
  }

  val simpleSource = Source(1 to 42)
  val simpleFlow = Flow[Int].map(x => x)
  val simpleSink = Sink.ignore
  val enhancedFlowCountFuture = simpleSource.viaMat(enhanceFlow(simpleFlow))(Keep.right).toMat(simpleSink)(Keep.left).run()
  enhancedFlowCountFuture.onComplete {
    case Success(count) => println(s"$count elements went through the enhanced flow")
    case _ => println("Something failed")
  }


}
